package cn.territory

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}
import spire.std.array
import spire.syntax.std.array

object territoryV1 {

  def main(args: Array[String]): Unit = {
    //校验参数
    if(args.length!=1){
      println(
        """
          |请输入参数:
          |inputpath
          |
        """.stripMargin)
      sys.exit()
    }
    val Array (inputpath) = args
    //读取本类名称 s"${this.getClass.getSimpleName}
    val conf: SparkConf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")

    val sc: SparkContext = new SparkContext(conf)

    val sqlcontext: SQLContext = new SQLContext(sc)
    //读取parquet文件
    val parquet: DataFrame = sqlcontext.read.parquet(inputpath)
    //构建视图
    parquet.registerTempTable("region")
    //sql查询
    val sql: DataFrame = sqlcontext.sql(
      """
      select
      provincename
      ,cityname
      ,sum (case when processnode >=1 and requestmode =1 then 1 else 0 end ) originlRequest
      ,sum (case when processnode >=2 and requestmode =1 then 1 else 0 end ) valueRequest
      ,sum (case when processnode =3 and requestmode =1 then 1 else 0 end ) advertiseRequest
      ,sum (case when iseffective = '1'and isbilling='1' and isbid='1' and adorderid !=0 then 1 else 0 end) biddingTime
      ,sum (case when iseffective = '1'and isbilling='1' and iswin ='1' then 1 else 0 end) biddingSucess
      ,sum (case when iseffective = '1' and processnode =2 then 1 else 0 end) terminalShow
      ,sum (case when iseffective = '1'and isbilling='1' and iswin ='1' then winprice/1000  else 0  end) DSPAdvertiseConsuem
      ,sum (case when iseffective = '1' and processnode =3 then 1 else 0 end) audienceClick
      |,sum (case when iseffective = '1'and isbilling='1' and iswin ='1' then adpayment/1000  else 0  end) DSPAdvertiseCos
      from region
      group by provincename,cityname
      """.stripMargin)

    sql.show()
    //加载配置文件,会加载application.conf,applicarion.json,application.properties文件
    //写入mysql
    val load: Config = ConfigFactory.load()
    val properties: Properties = new Properties()
    properties.put("user",load.getString("jdbc.user"))
    properties.put("password",load.getString("jdbc.password"))
    sql.write.mode(SaveMode.Ignore).jdbc(load.getString("jdbc.url"),load.getString("jdbc.tableNames"),
          properties)

    //判断路径是否存在
//    val hadoopConfiguration = sc.hadoopConfiguration
//    val fs = FileSystem.get(hadoopConfiguration)
//
//    val resultPath = new Path("D:\\logs\\cv")
//    if(fs.exists(resultPath)) {
//      fs.delete(resultPath, true)
//    }
//
//    //以json的形式写入文件
//    sql.coalesce(1).write.json("D:\\logs\\cv")

    sc.stop()

  }

}
